package com.zhang.gmall.app.func;

import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.common.GmallConfig;
import com.zhang.gmall.utils.DimUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Collection;
import java.util.Locale;
import java.util.Set;

/**
 * @title: 自定义写入Phoenix表sink
 * @author: zhang
 * @date: 2022/3/23 19:33
 * 根据数据流中的sinkTable信息动态写入不同的表中
 */
public class DimSinkFunc extends RichSinkFunction<JSONObject> {
    private Connection connection;

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
        //开启自动提交事务
        connection.setAutoCommit(true);
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        PreparedStatement statement = null;
        try {
            //获取类型
/*            String type = value.getString("type");
            if ("bootstrap-start".equals(type) || "bootstrap-complete".equals(type)) {
                return;
            }*/
            String sinkTable = value.getString("sinkTable");
            JSONObject data = value.getJSONObject("data");

            String upsertSql = genUpsertSql(data, sinkTable);
            System.out.println("upsert语句：" + upsertSql);
            statement = connection.prepareStatement(upsertSql);

            // 判断，如果是更新操作，则需要将redis中的缓存数据删除
            // 避免数据不一致问题
            String type = value.getString("type");
            if ("update".equals(type)){
                DimUtil.deleteCached(sinkTable.toUpperCase(),value.getString("id"));
            }

            statement.execute();
        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("向Phoenix表中插入数据失败！");
        } finally {
            if (statement != null) {
                statement.close();
            }
        }
    }

    //upsert into student(id,name,sex) values('1001','zhangsan','beijing');
    private String genUpsertSql(JSONObject data, String sinkTable) {
        Set<String> keySet = data.keySet();
        Collection<Object> values = data.values();
        StringBuilder upsertSQL = new StringBuilder("upsert into ");
        upsertSQL.append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append("( ")
                .append(org.apache.commons.lang.StringUtils.join(keySet, ","))
                .append(" ) ")
                .append("values('")
                .append(StringUtils.join(values, "','"))
                .append("')");
        return upsertSQL.toString();
    }

    @Override
    public void close() throws Exception {
        connection.close();
    }
}
